const fs = require("fs");
const EventEmitter = require("events");

class MyReadStream extends EventEmitter {
  constructor(path, options = {}) {
    super();
    this.path = path;
    this.flags = options.flags || "r";
    this.mode = options.mode || 438;
    this.autoClose = options.autoClose || true;
    this.start = options.start || 0;
    this.end = options.end;
    this.highWaterMark = options.highWaterMark || 64 * 1024;
    this.readOffset = 0;

    // 源码可读流只要一执行 new 操作，就会执行open操作
    this.open();
    // 监听新的事件后会被触发
    this.on("newListener", (type) => {
      if (type === "data") {
        this.read();
      }
    });
  }

  open() {
    // 原生open打开指定位置的文件
    fs.open(this.path, this.flags, this.mode, (err, fd) => {
      if (err) return this.emit("error", err);

      this.fd = fd;
      // 主动触发open事件，供外部使用
      this.emit("open", fd);
    });
  }

  read() {
    // 获取不到this.fd, 原因：这里获取this.fd是同步代码，而设置fd在open的回调函数中，属于异步代码， 解决方法： 发布订阅
    // console.log(this.fd);
    // once: 只执行一次
    if (typeof this.fd !== "number") {
      // 先执行open事件，在执行read事件，确保this.fd有值
      return this.once("open", this.read);
    }
    // data事件为异步，不需要设置open与read的执行顺序
    let buf = Buffer.alloc(this.highWaterMark); // 缓存空间
    let howMuchToRead = this.end
      ? Math.min(this.end - this.readOffset + 1, this.highWaterMark)
      : this.highWaterMark;
    /*
      if (this.end) {
        howMuchToRead = Math.min(
          this.end - this.readOffset + 1,
          this.highWaterMark
        );
      } else {
        // 初始化时没有设置end值
        howMuchToRead = this.highWaterMark;
      }
    */

    fs.read(
      this.fd,
      buf,
      0,
      howMuchToRead,
      this.readOffset,
      (err, readBytes) => {
        if (readBytes) {
          this.readOffset += readBytes;
          this.emit("data", buf.subarray(0, readBytes));
          this.read();
        } else {
          this.emit("end");
          this.close();
        }
      }
    );
  }

  close() {
    fs.close(this.fd, () => {
      this.emit("close");
    });
  }

  pipe(ws) {
    this.on("data", (chunk) => {
      let flag = ws.write(chunk);
      if (!flag) {
        // 未实现
        this.pause();
      }
    });

    ws.on("drain", () => {
      // 未实现·
      this.resume();
    });
  }
}

module.exports = MyReadStream;
